package com.caissa.springboot.starter.kafka.autoconfigure;

import com.base.springboot.starter.kafka.consumer.errorhandle.ErrMsg2EsHandler;
import com.base.springboot.starter.kafka.consumer.retry.DefaultBackOffRetryTemplate;
import com.base.springboot.starter.kafka.service.KafkaService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;

/**
 * @author yuzhongo
 * @date 2020/8/17 下午3:48
 */
@Configuration
@Slf4j
@EnableConfigurationProperties(KafkaProperties.class)
@ConditionalOnClass(KafkaTemplate.class)
public class KafkaAutoConfiguration {

    @Bean
    @ConditionalOnClass
    public KafkaService kafkaUtil() {
        return new KafkaService();
    }

    /**
     * 创建kafka监听器自定义工厂，实现自动错误消息转存和自动ack offset的功能
     * 使用方法： 在@listener的 containerFactory = "caissaIdempoListenerContainerFactory"
     *
     * 提交策略：每执行一条消息
     *  成功，自动提交ack；
     *  异常，直接写ES后，自动ack
     *
     * @param configurer
     * @return
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory caissaIdempoListenerContainerFactory(
            ConsumerFactory consumerFactory,
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer) {

        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();

        configurer.configure(factory, consumerFactory);

        //由于重试在应用层由开发同学自己控制更加灵活，且这里的重试会同步阻塞消费线程，异步处理也会带来额外流程，目前考虑做省事儿的办法是去
        //factory.setRetryTemplate(new DefaultBackOffRetryTemplate());

        factory.setErrorHandler(new ErrMsg2EsHandler("caissaDefault"));

        ContainerProperties containerProperties = factory.getContainerProperties();

        //每个消息执行ack
        containerProperties.setAckMode(ContainerProperties.AckMode.RECORD);

        return factory;
    }
}